package com.air.antispider.stream.dataprocess.businessprocess

import java.lang

import com.air.antispider.stream.common.util.jedis.{JedisConnectionUtil, PropertiesUtil}
import com.air.antispider.stream.common.util.spark.SparkMetricsUtils
import com.alibaba.fastjson.JSONObject
import org.apache.commons.lang3.time.FastDateFormat
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.json4s.DefaultFormats
import org.json4s.jackson.Json
import redis.clients.jedis.JedisCluster

/**
  * Spark性能监控
  */
object SparkStreamingMonitor {

  /**
    * 数据预处理模块性能监控
    *
    * @param serverCountRDD
    */
  def streamMonitor(serverCountRDD: RDD[(String, Int)], sc: SparkContext) = {
    //    1.  能获取到Spark监控数据
    //两种方式:
    // 1.使用代码直接访问http://localhost:4040/metrics/json/获取数据
    // 2. 使用Spark提供的工具类来获取

    /*
//如果放到运行环境,获取主机地址需要用下面的代码来获取
val sparkDriverHost =
sc.getConf.get("spark.org.apache.hadoop.yarn.server.webproxy.amfilter.AmIpFilter.param.PROXY _URI_BASES")
//在 yarn 上运行的监控数据 json 路径
val url = s"${sparkDriverHost}/metrics/json" */

    val url = "http://localhost:4040/metrics/json/"
    //调用Spark提供的工具类来获取监控数据
    val jsonObj: JSONObject = SparkMetricsUtils.getMetricsJson(url)
    //      2.  得到计算开始时间/结束时间
    //2.1获取gauges节点数据
    val gaugesObj: JSONObject = jsonObj.getJSONObject("gauges")
    //2.2 获取开始时间对象/结束时间对象
    //local-应用ID.driver.应用名称.StreamingMetrics.streaming.lastCompletedBatch_processingStartTime
    //应用ID
    val applicationId: String = sc.applicationId
    //应用名称
    val appName: String = sc.appName
    //创建开始时间的Key
    val startTimeName = applicationId + ".driver." + appName + ".StreamingMetrics.streaming.lastCompletedBatch_processingStartTime"
    //创建开始时间的Key
    val endTimeName = applicationId + ".driver." + appName + ".StreamingMetrics.streaming.lastCompletedBatch_processingEndTime"

    val startObj: JSONObject = gaugesObj.getJSONObject(startTimeName)
    //获取开始时间
    val startTime: lang.Long = startObj.getLong("value")
    //获取结束时间
    val endTime: lang.Long = gaugesObj.getJSONObject(endTimeName).getLong("value")

    //    jsonObj.getJSONObject("gauges").getJSONObject(startTimeName).getLong("value")

    //      3.  得到计算的链路采集数量
    val count: Long = serverCountRDD.count()
    //      4.  计算数据采集速度
    //计算时间差
    val costTime = endTime - startTime
    //使用  数量/时间差=速度
    var countPerMillis = 0.0
    if (costTime != 0){
      //如果costTime 不等于0 ,才相除
      countPerMillis = count.toDouble / costTime.toDouble
    }


    //      5.  将最终的数据放入Redis,让前端展示
    //通过跟踪web项目的代码,发现了JsonVO对象,那么接下来将数据封装为JsonVO对应的格式
    val serversCountMap: collection.Map[String, Int] = serverCountRDD.collectAsMap()

    //因为web项目需要的最后时间是yyyy-MM-dd HH:mm:ss hh是12小时制,HH24小时制
    val endTimeStr: String = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss").format(endTime)
    val map = Map[String, Any](
      "costTime" -> costTime,
      "serversCountMap" -> serversCountMap,
      "applicationId" -> applicationId,
      "countPerMillis" -> countPerMillis,
      "applicationUniqueName" -> appName,
      "endTime" -> endTimeStr,
      "sourceCount" -> count
    )

    //将map转换为JSON数据
    val json: String = Json(DefaultFormats).write(map)
    //将数据放入Redis

    val jedis: JedisCluster = JedisConnectionUtil.getJedisCluster
    //存入Redis的key
    val key = PropertiesUtil.getStringByKey("cluster.key.monitor.dataProcess", "jedisConfig.properties") + System.currentTimeMillis()
    //超时时间
    val ex = PropertiesUtil.getStringByKey("cluster.exptime.monitor", "jedisConfig.properties").toInt
    jedis.setex(key,ex,json)

  }

}
